/*
 * Copyright (c) 2022. China Mobile (SuZhou) Software Technology Co.,Ltd. All rights reserved.
 * Lakehouse is licensed under Mulan PSL v2.
 * You can use this software according to the terms and conditions of the Mulan PSL v2.
 * You may obtain a copy of Mulan PSL v2 at:
 *          http://license.coscl.org.cn/MulanPSL2
 * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
 * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
 * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
 * See the Mulan PSL v2 for more details.
 */

package com.chinamobile.cmss.lakehouse.engine.spark;

import com.chinamobile.cmss.lakehouse.common.engine.CommandBuilder;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;

public class SparkCommandBuilder extends CommandBuilder<SparkCommandBuilder> {

    private boolean verbose;
    private String appName;
    private String deployMode;
    private String mainClass;
    private String master;
    private String proxyUser;

    public SparkCommandBuilder() {
        super();
    }

    public SparkCommandBuilder master(String master) {
        this.master = master;
        return self();
    }

    public SparkCommandBuilder verbose(boolean verbose) {
        this.verbose = verbose;
        return self();
    }

    public String getAppName() {
        return this.appName;
    }

    public SparkCommandBuilder appName(String appName) {
        this.appName = appName;
        return self();
    }

    public SparkCommandBuilder setDeployMode(String deployMode) {
        this.deployMode = deployMode;
        return self();
    }

    public SparkCommandBuilder setMainClass(String mainClass) {
        this.mainClass = mainClass;
        return self();
    }

    public SparkCommandBuilder setProxyUser(String proxyUser) {
        this.proxyUser = proxyUser;
        return self();
    }

    @Override
    public List<String> buildSubmitArgs() {
        List<String> args = new ArrayList<>();

        if (verbose) {
            args.add("--verbose");
        }

        if (master != null) {
            args.add("--master");
            args.add(master);
        }

        if (deployMode != null) {
            args.add("--deploy-mode");
            args.add(deployMode);
        }

        if (appName != null) {
            args.add("--name");
            args.add(appName);
        }

        for (Map.Entry<String, String> e : conf.entrySet()) {
            args.add("--conf");
            args.add(String.format("%s=%s", e.getKey(), e.getValue()));
        }

        if (mainClass != null) {
            args.add("--class");
            args.add(mainClass);
        }

        if (proxyUser != null) {
            args.add("--proxy-user");
            args.add(proxyUser);
        }

        if (appResource != null) {
            args.add(appResource);
        }

        args.addAll(appArgs);

        return args;
    }

    @Override
    public List<String> buildKillArgs(String podId) {
        List<String> args = new ArrayList<>();

        args.add("--kill");
        args.add(podId);

        if (master != null) {
            args.add("--master");
            args.add(master);
        }

        for (Map.Entry<String, String> e : conf.entrySet()) {
            args.add("--conf");
            args.add(String.format("%s=%s", e.getKey(), e.getValue()));
        }
        return args;
    }

    @Override
    protected SparkCommandBuilder self() {
        return this;
    }
}
